package com.nyist.mq.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.nyist.autoconfigure.RocketMQConfigProperties;
import com.nyist.mq.service.ProducerService;
import com.nyist.mq.constant.RocketTagConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

/**
 * 1.
 *
 * @author yangxiaogen
 * @date 2023/3/21 16:51
 */
@Slf4j
@Service
public class ProducerServiceImpl implements ProducerService {

    private static String topic;
    private static Integer sendMessageTimeout;

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Resource
    private RocketMQConfigProperties rocketMQConfigProperties;

    @PostConstruct
    public void init() {
        topic = rocketMQConfigProperties.getTopic();
        sendMessageTimeout = rocketMQConfigProperties.getSendMessageTimeout();
    }

    /**
     * 无返回发送消息 , 只管发送
     *
     * @param message
     * @return void
     * @author yangxiaogen
     * @date 2023/3/23 15:58
     */
    @Override
    public void send(String message) {
        log.info("=====>>>send,准备发往rocketmq的消息为:{}", message);
        rocketMQTemplate.convertAndSend(topic + ":" + RocketTagConstant.TAG_SEND, message);
    }

    /**
     * 发送同步消息（阻塞当前线程，等待broker响应发送结果，这样不太容易丢失消息）
     *
     * @param message
     * @return org.apache.rocketmq.client.producer.SendResult
     * @author yangxiaogen
     * @date 2023/3/23 16:07
     */
    @Override
    public SendResult syncSend(String message) {
        // 这种处理 message 的方式上面直接使用 convert的方式相同
        log.info("=====>>>syncSend,准备发往rocketmq的消息为:{}", message);
        SendResult sendResult = rocketMQTemplate.syncSend(topic + ":" + RocketTagConstant.TAG_SYNC_SEND, MessageBuilder.withPayload(message).build());
        log.info("=====>>>result,rocketmq响应信息为:{}", JSONObject.toJSONString(sendResult));
        return sendResult;
    }

    /**
     * （通过线程池执行发送到 broker 的消息任务，执行完后回调：在SendCallback中可处理相关 成功 异常 时的逻辑）
     * 发送异步消息 (适用于响应时间敏感业务场景)
     * 通过 sendCallback 来获取发送消息, onSuccess 发送成功 . onException 发送异常
     *
     * @param message
     * @return org.apache.rocketmq.client.producer.SendResult
     * @author yangxiaogen
     * @date 2023/3/23 16:07
     */
    @Override
    public void asyncSend(String message) {
        // 这种处理 message 的方式上面直接使用 convert的方式相同
        log.info("=====>>>asyncSend,异步发送准备发往rocketmq的消息为:{}", message);
        rocketMQTemplate.asyncSend(topic + ":" + RocketTagConstant.TAG_ASYNC_SEND, MessageBuilder.withPayload(message).build(), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("=====>>>result,rocketmq响应信息为success:{}", JSONObject.toJSONString(sendResult));
            }

            @Override
            public void onException(Throwable e) {
                log.error("=====>>>result,rocketmq响应信息为error:{}", JSONObject.toJSONString(e.getMessage()));
            }
        });
    }

    /**
     * 发送延时消息（上面的发送同步消息，delayLevel的值就为0，因为不延时）
     * 在start版本中 延时消息一共分为18个等级分别为：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     */
    @Override
    public void sendDelayMsg(String message, int delayLevel) {
        log.info("=====>>>延时,等级:{},无tag,同步发送,准备发往rocketmq的消息为:{}",delayLevel, message);
        rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(message).build(), sendMessageTimeout, delayLevel);
    }

    /**
     * 发送单向消息（只负责发送消息，不等待应答，不关心发送结果，如日志）
     */
    @Override
    public void sendOneWayMsg(String message) {
        log.info("=====>>>无===tag,单向发送准备发往rocketmq的消息为:{}", message);
        rocketMQTemplate.sendOneWay(topic, MessageBuilder.withPayload(message).build());
    }

    /**
     * 发送带tag的消息，直接在topic后面加上":tag"
     */
    @Override
    public SendResult sendTagMsg(String message, String tag) {
        log.info("=====>>>input===tag={},同步发送准备发往rocketmq的消息为:{}",tag, message);
        return rocketMQTemplate.syncSend(topic + ":" + tag, MessageBuilder.withPayload(message).build());
    }

    @Override
    public String testUser() {
        return topic + "::" + sendMessageTimeout;
    }

}
